package test;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.Cell;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class Text {
	public static Configuration configuration;
	public static Connection connection;
	public static Admin admin;

	public static void init() {
		configuration = HBaseConfiguration.create();
		configuration.set("hbase.rootdir", "hdfs://localhost:9000/hbase");
		try {
			connection = ConnectionFactory.createConnection(configuration);
			admin = connection.getAdmin();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void close() {
		try {
			if (admin != null) {
				admin.close();
			}
			if (connection != null) {
				connection.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void createTable(String tableName, String[] fields) throws IOException {
		init();
		TableName tablename = TableName.valueOf(tableName);
		if (admin.tableExists(tablename)) {
			System.out.println("table is exists!");
			admin.disableTable(tablename);
			admin.deleteTable(tablename);
		}
		TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tablename);
		for (int i = 0; i < fields.length; i++) {
			ColumnFamilyDescriptor family = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(fields[i])).build();
			tableDescriptor.setColumnFamily(family);
		}
		admin.createTable(tableDescriptor.build());
		close();
	}

	public static void addRecord(String tableName, String row, String[] fields, String[] values) throws IOException {
		init();
		Table table = connection.getTable(TableName.valueOf(tableName));
		Put put = new Put(row.getBytes());
		for (int i = 0; i < fields.length; i++) {
			String[] cols = fields[i].split(":");
			if (cols.length == 1) {
				put.addColumn(fields[i].getBytes(), "".getBytes(), values[i].getBytes());
			} else {
				put.addColumn(cols[0].getBytes(), cols[1].getBytes(), values[i].getBytes());
			}
			table.put(put);
		}
		close();
	}

	public static void showResult(Result result) {
		Cell[] cells = result.rawCells();
		for (int i = 0; i < cells.length; i++) {
			System.out.println("RowName:" + new String(CellUtil.cloneRow(cells[i])));// 打印行键
			System.out.println("ColumnName:" + new String(CellUtil.cloneQualifier(cells[i])));// 打印列名
			System.out.println("Value:" + new String(CellUtil.cloneValue(cells[i])));// 打印值
			System.out.println("Column Family:" + new String(CellUtil.cloneFamily(cells[i])));// 打印列簇
			System.out.println();
		}
	}

	public static void scanColumn(String tableName, String column) {
		init();
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.addFamily(Bytes.toBytes(column));
			ResultScanner scanner = table.getScanner(scan);
			for (Result result = scanner.next(); result != null; result = scanner.next()) {
				showResult(result);
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			close();
		}
	}

	public static void modifyData(String tableName, String row, String column, String value) throws IOException {
		init();
		Table table = connection.getTable(TableName.valueOf(tableName));
		Put put = new Put(row.getBytes());
		String[] cols = column.split(":");
		if (cols.length == 1) {
			put.addColumn(column.getBytes(), "".getBytes(), value.getBytes());
		} else {
			put.addColumn(cols[0].getBytes(), cols[1].getBytes(), value.getBytes());
		}
		table.put(put);
		close();
	}

	public static void deleteRow(String tableName, String row) throws IOException {
		init();
		Table table = connection.getTable(TableName.valueOf(tableName));
		Delete delete = new Delete(row.getBytes());
		table.delete(delete);
		close();
	}
//	1
	public static void main(String[] args) {
		String[] fields = {"id","score"};
		try{
			createTable("test",fields);
		}catch(IOException e){
			e.printStackTrace();
		}
	}
//	2
//	public static void main(String[] args){
//		String[] fields = {"Score:Math","Score:Computer Science","Score:English"};
//		String[] values = {"90","90","90"};
//		try{
//			addRecord("grade","S_Name",fields,values);
//		}catch(IOException e){
//			e.printStackTrace();
//		}
//	}
//  3	
//	public static void main(String[] args){
//		scanColumn("Student","S_Age");
//	}
//  4
//	public static void main(String[] args){
//		try{
//			modifyData("Student","1","S_Name","Tom");
//		}
//		catch(Exception e){
//			e.printStackTrace();
//		}
//	}
// 5
//	public static void main(String[] args){
//		try{
//			deleteRow("Student","3");
//		}catch(Exception e){
//			e.printStackTrace();
//		}
//	}
}
